package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo01WordCount {
  def main(args: Array[String]): Unit = {

    // 初始化Spark环境

    // 创建Spark配置对象
    val conf: SparkConf = new SparkConf()
    // 配置任务的名称
    conf.setAppName("Demo01WordCount")
    // 配置Spark的运行方式 local表示本地运行
    conf.setMaster("local")

    // 1、创建Spark上下文对象 即 Spark的入口
    val sc: SparkContext = new SparkContext(conf)

    // 2、读取文件 将文件构建成RDD
    /**
     * RDD：弹性分布式数据集
     * ”暂时”可以将其理解为Scala中的一个集合，使用起来类似List
     * 实际上RDD是Spark中提供的一个编程模型
     *
     * Spark并没有自己的读文件的方式 使用的还是MR的切片、格式化数据的方式 是基于TextInputFormat类
     */
    val linesRDD: RDD[String] = sc.textFile("Spark/data/wordCnt/input")

    // 3、对每行数据进行切分 并对切分后的结果进行扁平化处理
    // flatMap等其他跟RDD相关的方法在Spark中统称为“算子”
    val wordsRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))

    // 4、按照单词进行分组
    /**
     * Iterable迭代器跟List区别？
     * 1、迭代器只能被遍历一次
     * 2、迭代器的数据不在内存中 当使用时才会逐条加载到内存里 List的数据都在内存中
     */
    val groupRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)

    // 5、统计每个单词的数量 并整理最后输出的结果
    val wordCntRDD: RDD[String] = groupRDD.map(kv => s"${kv._1},${kv._2.size}")

    // 6、打印或保存最后的结果
    //    wordCntRDD.foreach(println)
    wordCntRDD.saveAsTextFile("Spark/data/wordCnt/output")

    // 防止Spark程序停止 WebUI会被销毁
    while(true){

    }

    /**
     * 1、搭建Spark On Yarn模式 并以两种方式提交Spark自带的SparkPi任务 将最终输出的Pi的结果截图
     * 2、（思考）如何将自己写好的WordCount程序提交到Yarn上运行
     */

  }

}
